package trunk.hadoop;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;

import java.io.*;
import java.util.ArrayList;

public class daMapper extends MapReduceBase implements Mapper<IntWritable, BytesWritable, Text, IntWritable>{
    private static ArrayList<queryWrapper> query = new ArrayList<>(0);

    public void configure(JobConf job){
        try {
            BufferedReader reader = new BufferedReader(new InputStreamReader( new FileInputStream(new File(DistributedCache.getCacheFiles(job)[0].getPath()))));
            String line;    int i = 0;
            while((line = reader.readLine())!=null)
            {
                query.add(new queryWrapper(i++,line.getBytes()));
            }


        } catch (IOException e){ e.printStackTrace(); }
    }
    @Override
    public void map(IntWritable keyIn, BytesWritable valueIn, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
        byte[] input = valueIn.getBytes();  int[] overlap;  byte[] pattern; int blockStart = keyIn.get();
        Text keyOut = new Text();           IntWritable valueOut = new IntWritable();   int patternLength;
        String currentFile = ((FileSplit)reporter.getInputSplit()).getPath().getName();

        for (queryWrapper currentQuery : query) {
            keyOut.set(currentFile + "," + currentQuery.toString());
            overlap = currentQuery.getOverlap();    pattern = currentQuery.getPattern();

            output.collect(keyOut, new IntWritable(-1)); // try make every query appear

            // m = start of potential match     // i = pos away from m      //
            int root = 0,     offset = 0;
            patternLength = pattern.length - 1;
            while (root + offset != input.length-1){
                if (pattern[offset] == input[(root + offset)])       // if char at 'i' = char in query [base loc + i]
                {
                    if (offset == patternLength)  // if exact match
                    {
                        valueOut.set(blockStart + root);    // adjust + 1 here if needed
                        output.collect(keyOut,valueOut);
                        root = (root + offset) - overlap[offset];
                        offset = 0;
                    }
                    else                        // if kinda gonna match maybe?
                        offset++;
                }
                else
                {
                    root = root + offset - overlap[offset];     // start pos + cur pos - overlap = will equal to real current pos in array
                    if (overlap[offset] != -1)        // check if not at the start of overlap string
                        offset = overlap[offset];
                    else
                        offset = 0;
                }
            }

        }
    }
}
